package org.lqq.serverRegisterAndFinder.etcd;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ning.http.client.*;
import mousio.etcd4j.EtcdClient;
import mousio.etcd4j.promises.EtcdResponsePromise;
import mousio.etcd4j.responses.EtcdErrorCode;
import mousio.etcd4j.responses.EtcdException;
import mousio.etcd4j.responses.EtcdKeysResponse;
import org.lqq.entity.DefaultServerAddress;
import org.lqq.entity.ServerAddress;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * Created by 李清泉 on 2017/11/2.
 *
 * @remark
 */
public class EtcdRefreshThread implements Runnable {

    private EtcdClient etcdClient;

    private String address;

    private String serverName;

    public EtcdRefreshThread(EtcdClient etcdClient,String serverName,String address){
        this.etcdClient=etcdClient;
        this.address=address;
        this.serverName=serverName;
    }
    @Override
    public void run() {
        try {
//            EtcdResponsePromise<EtcdKeysResponse> promise=etcdClient.getDir(EtcdConst.etcdKey+"/"+EtcdConst.sysKey).recursive().send();
//            EtcdKeysResponse response=promise.get();
//            if(response!=null){
//                createData(response);
//            }
//            EtcdResponsePromise<EtcdKeysResponse> p=etcdClient.getDir(EtcdConst.etcdKey+"/"+EtcdConst.sysKey).recursive().waitForChange().timeout(-1, TimeUnit.SECONDS).send();
//            p.addListener(promise1->{
//                Throwable t = promise1.getException();
//                if(t instanceof EtcdException){
//                    if(((EtcdException) t).isErrorCode(EtcdErrorCode.NodeExist)){
//                        System.out.println("EtcdErrorCode");
//                    }
//                }
//                // getNow() returns null on exception
//                EtcdKeysResponse r = promise1.getNow();
//                if(r != null){
//                    System.out.println(r.getAction().name());
//                   createData(r);
//                }
//            });
//            EtcdResponsePromise<EtcdKeysResponse> pr=etcdClient.get(EtcdConst.etcdKey+"/"+EtcdConst.sysKey).recursive().waitForChange().timeout(-1, TimeUnit.SECONDS).send();
//            pr.addListener(promise->{
//                EtcdKeysResponse r = promise.getNow();
//            });

            AsyncHttpClient asyncHttpClient = new AsyncHttpClient();
            asyncHttpClient.prepareGet("http://127.0.0.1:2379/v2/keys/serverList/test?wait=true&recursive=true&stream=true").setRequestTimeout(-1).execute(new AsyncHandler<String>() {

                @Override
                public void onThrowable(Throwable throwable) {
                    throwable.printStackTrace();
                }

                @Override
                public STATE onBodyPartReceived(HttpResponseBodyPart httpResponseBodyPart) throws Exception {
                    byte[] bytes=httpResponseBodyPart.getBodyPartBytes();
                    String content =new String(bytes);
                    if(content!=null&&content.length()>0) {
                        ObjectMapper objectMapper = new ObjectMapper();
                        EtcdReponseMessage message = objectMapper.readValue(content, EtcdReponseMessage.class);
                        if (message != null) {
                            System.out.println(message.getAction());
                            System.out.println(message.getNode().getKey());
                            if(!"update".equals(message.getAction())){
                                String key=message.getNode().getKey();
                                if(key!=null&&key.length()>0){
                                    int index=key.lastIndexOf("-");
                                    String ipaddr=key.substring(index+1,key.length());
                                    System.out.println("ipaddr:"+ipaddr);
                                    int i=key.lastIndexOf("/");
                                    String serverName=key.substring(i+1,index);
                                    System.out.println("serverName:"+serverName);
                                    int in=ipaddr.lastIndexOf(":");
                                    String port=ipaddr.substring(in+1,ipaddr.length());
                                    String ip=ipaddr.substring(0,in);
                                    ServerAddress serverAddress=new DefaultServerAddress(serverName,Integer.parseInt(port),ip);
                                    EtcdData.getInstance().put(serverName,serverAddress);
                                }
                            }
                        }
                    }
                    return STATE.CONTINUE;
                }

                @Override
                public STATE onStatusReceived(HttpResponseStatus httpResponseStatus) throws Exception {
                    return STATE.CONTINUE;
                }

                @Override
                public STATE onHeadersReceived(HttpResponseHeaders httpResponseHeaders) throws Exception {
                    return STATE.CONTINUE;
                }

                @Override
                public String onCompleted() throws Exception {
                    return "complete!";
                }
            });
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void createData(EtcdKeysResponse response) {
        EtcdKeysResponse.EtcdNode node= response.getNode();
        List<ServerAddress> list=new ArrayList<>();
        if(node.isDir()){
            List<EtcdKeysResponse.EtcdNode> nodes=node.getNodes();
            for(EtcdKeysResponse.EtcdNode n:nodes){
                ServerAddress serverAddress=trans(n.getValue());
                list.add(serverAddress);
            }
        }else {
             String value=node.getValue();
             list.add(trans(value));
        }
        Map<String,Set<ServerAddress>> map=new HashMap<>();
        for(ServerAddress serverAddress:list){
            String name=serverAddress.getName();
            Set<ServerAddress> serverAddresses=map.get(name);
            if(serverAddresses==null){
                serverAddresses=new HashSet<>();
                map.put(name,serverAddresses);
            }
            serverAddresses.add(serverAddress);
        }
        for(String key:map.keySet()){
            Set<ServerAddress> serverAddresses=map.get(key);
            EtcdData.getInstance().put(key,serverAddresses);
            for (ServerAddress serverAddress:serverAddresses){
                System.out.println("put:"+serverAddress);
            }
        }
    }

    private ServerAddress trans(String value){
        String[] vals=value.split("-");
        String[] f=vals[1].split(":");
        ServerAddress serverAddress=new DefaultServerAddress(vals[0],Integer.parseInt(f[1]),f[0]);
        return serverAddress;
    }
}
